package de.ipk_gatersleben.bit.bi.edal.primary_data;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;
import java.util.StringTokenizer;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.zip.ZipOutputStream;

import javax.mail.internet.AddressException;
import javax.mail.internet.InternetAddress;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.handler.AbstractHandler;

import de.ipk_gatersleben.bit.bi.edal.primary_data.file.EdalException;
import de.ipk_gatersleben.bit.bi.edal.primary_data.file.PrimaryDataDirectory;
import de.ipk_gatersleben.bit.bi.edal.primary_data.file.PrimaryDataEntity;
import de.ipk_gatersleben.bit.bi.edal.primary_data.file.PrimaryDataEntityVersionException;
import de.ipk_gatersleben.bit.bi.edal.primary_data.file.PrimaryDataFile;
import de.ipk_gatersleben.bit.bi.edal.primary_data.file.PrimaryDataFileException;
import de.ipk_gatersleben.bit.bi.edal.primary_data.metadata.DataFormat;
import de.ipk_gatersleben.bit.bi.edal.primary_data.metadata.DataSize;
import de.ipk_gatersleben.bit.bi.edal.primary_data.metadata.EnumDublinCoreElements;
import de.ipk_gatersleben.bit.bi.edal.primary_data.metadata.MetaDataException;
import de.ipk_gatersleben.bit.bi.edal.primary_data.reference.EdalApprovalException;
import de.ipk_gatersleben.bit.bi.edal.primary_data.reference.PersistentIdentifier;

public class EdalJettyHandler extends AbstractHandler {

	protected static final String NOT_FOUND = "Not Found";
	protected static final int OK_STATUS = 200;
	protected static final int NOT_FOUND_STATUS = 404;

	private static HashMap<Integer, List<String>> hashmap = new HashMap<Integer, List<String>>();

	private static ThreadPoolExecutor executor;

	static {
		executor = new EdalThreadPoolExcecutor(1, Math.max(1, Runtime
				.getRuntime().availableProcessors()), 10, TimeUnit.SECONDS,
				new ArrayBlockingQueue<Runnable>(10));
	}

	@Override
	public void handle(String target, Request baseRequest,
			HttpServletRequest request, HttpServletResponse response)
			throws IOException, ServletException {

		if (request.getMethod().equalsIgnoreCase("GET")) {

			final String url = request.getRequestURI().toString();

			DataManager.getImplProv().getLogger().debug(url);

			final StringTokenizer tokenizer = new StringTokenizer(url,
					EdalJettyServer.EDAL_PATH_SEPARATOR);

			if (!tokenizer.hasMoreTokens()) {
				/* no method defined -> break */
				this.sendMessage(response, EdalJettyHandler.NOT_FOUND_STATUS,
						EdalJettyHandler.NOT_FOUND,
						"no ID and version specified");
			} else {
				final String methodToken = tokenizer.nextToken().toUpperCase();

				try {
					switch (EdalHttpFunctions.valueOf(methodToken)) {

					case DOI:
					case URL:
						if (tokenizer.hasMoreTokens()) {
							final String internalId = tokenizer.nextToken();

							if (tokenizer.hasMoreTokens()) {

								final String uuidToken = tokenizer.nextToken();

								if (tokenizer.hasMoreTokens()) {
									final String versionToken = tokenizer
											.nextToken();
									try {
										final long versionNumber = Long
												.parseLong(versionToken);

										PrimaryDataEntity entity = null;

										/** code for reviewer landing page */
										if (tokenizer.hasMoreTokens()) {
											final String reviewerToken = tokenizer
													.nextToken();

											if (tokenizer.hasMoreElements()) {
												final String methodTokenForReviewer = tokenizer
														.nextToken();

												/** process for reviewer : */

												if (methodTokenForReviewer
														.equalsIgnoreCase(EdalHttpFunctions.DOWNLOAD
																.name())) {

													this.responseDownloadRequest(
															response, entity,
															uuidToken,
															versionNumber,
															internalId,
															methodToken,
															reviewerToken);
													break;

												} else if (methodTokenForReviewer
														.equalsIgnoreCase(EdalHttpFunctions.ZIP
																.name())) {

													this.responseZipRequest(
															response, entity,
															uuidToken,
															versionNumber,
															internalId,
															methodToken,
															reviewerToken);
													break;

												} else {

													this.sendMessage(
															response,
															EdalJettyHandler.NOT_FOUND_STATUS,
															EdalJettyHandler.NOT_FOUND,
															"unable to process '"
																	+ methodTokenForReviewer
																	+ "'");
													break;
												}

											} else {
												/**
												 * check if request is for
												 * DOWNLOAD function
												 */
												if (reviewerToken
														.equalsIgnoreCase(EdalHttpFunctions.DOWNLOAD
																.name())) {

													this.responseDownloadRequest(
															response, entity,
															uuidToken,
															versionNumber,
															internalId,
															methodToken, null);

												}
												/**
												 * check if request is for ZIP
												 * function function
												 */
												else if (reviewerToken
														.equalsIgnoreCase(EdalHttpFunctions.ZIP
																.name())) {

													this.responseZipRequest(
															response, entity,
															uuidToken,
															versionNumber,
															internalId,
															methodToken, null);
													break;

												}

												else {
													/**
													 * check if request is a
													 * reviewer code
													 */
													int reviewer = 0;
													try {
														reviewer = Integer
																.parseInt(reviewerToken);
													} catch (final NumberFormatException e) {
														this.sendMessage(
																response,
																EdalJettyHandler.NOT_FOUND_STATUS,
																EdalJettyHandler.NOT_FOUND,
																"unable to process reviewer ID '"
																		+ reviewerToken
																		+ "', please check again");
														break;
													}

													/**
													 * check if reviewer code
													 * exists
													 */
													entity = DataManager
															.getPrimaryDataEntityForReviewer(
																	uuidToken,
																	versionNumber,
																	internalId,
																	reviewer);
													this.sendEntityMetaDataForReviewer(
															response,
															entity,
															versionNumber,
															internalId,
															PersistentIdentifier
																	.valueOf(methodToken),
															reviewer);
													break;
												}
											}
										}
										/** end code for reviewer landing page */
										else {

											entity = DataManager
													.getPrimaryDataEntityForPersistenIdentifier(
															uuidToken,
															versionNumber,
															PersistentIdentifier
																	.valueOf(methodToken));
											this.sendEntityMetaDataForPersistentIdentifier(
													response,
													entity,
													versionNumber,
													PersistentIdentifier
															.valueOf(methodToken),
													internalId);
											break;
										}

									} catch (final EdalException e) {
										this.sendMessage(
												response,
												EdalJettyHandler.NOT_FOUND_STATUS,
												EdalJettyHandler.NOT_FOUND,
												"unable to send data : "
														+ e.getMessage());
									}
								} else {
									this.sendMessage(response,
											EdalJettyHandler.NOT_FOUND_STATUS,
											EdalJettyHandler.NOT_FOUND,
											"no version number set");
								}
							} else {
								this.sendMessage(response,
										EdalJettyHandler.NOT_FOUND_STATUS,
										EdalJettyHandler.NOT_FOUND,
										"no entity id set");
							}
						} else {
							this.sendMessage(response,
									EdalJettyHandler.NOT_FOUND_STATUS,
									EdalJettyHandler.NOT_FOUND,
									"no internal id set");
						}
						break;

					/* request an object */
					case EDAL:

						final String uuidToken = tokenizer.nextToken();

						if (tokenizer.hasMoreTokens()) {
							final String versionToken = tokenizer.nextToken();
							try {
								final int versionNumber = Integer
										.parseInt(versionToken);
								PrimaryDataEntity entity = null;

								if (tokenizer.hasMoreElements()) {

									final String downloadToken = tokenizer
											.nextToken();
									if (EdalHttpFunctions.valueOf(
											downloadToken.toUpperCase())
											.equals(EdalHttpFunctions.DOWNLOAD)) {

										entity = DataManager
												.getPrimaryDataEntityByID(
														uuidToken,
														versionNumber);

										if (!entity.isDirectory()) {
											this.sendFile(
													(PrimaryDataFile) entity,
													versionNumber, response);
										} else {
											this.sendEntityMetaData(response,
													entity);
										}
									}
								} else {
									entity = DataManager
											.getPrimaryDataEntityByID(
													uuidToken, versionNumber);

									this.sendEntityMetaData(response, entity);
								}

							} catch (final NumberFormatException
									| EdalException e) {
								if (e.getClass().equals(
										NumberFormatException.class)) {
									this.sendMessage(response,
											EdalJettyHandler.NOT_FOUND_STATUS,
											EdalJettyHandler.NOT_FOUND,
											"unable to cast '" + versionToken
													+ "' to a version number");
								} else {
									this.sendMessage(
											response,
											EdalJettyHandler.NOT_FOUND_STATUS,
											EdalJettyHandler.NOT_FOUND,
											"unable to send data : "
													+ e.getMessage());
								}
							}
						} else {
							this.sendMessage(response,
									EdalJettyHandler.NOT_FOUND_STATUS,
									EdalJettyHandler.NOT_FOUND,
									"no version number set");
						}
						break;

					/* accept an object */
					case ACCEPT:

						final String ticketAccept = tokenizer.nextToken();

						if (tokenizer.hasMoreElements()) {

							try {
								final int reviewerHashCode = Integer
										.parseInt(tokenizer.nextToken());

								if (checkIfAllowedToClick(reviewerHashCode,
										ticketAccept)) {

									DataManager
											.getImplProv()
											.getApprovalServiceProvider()
											.newInstance()
											.accept(ticketAccept,
													reviewerHashCode);
								} else {
									this.sendMessage(response,
											EdalJettyHandler.NOT_FOUND_STATUS,
											EdalJettyHandler.NOT_FOUND, "Already clicked");
									break;
								}
							} catch (EdalApprovalException
									| InstantiationException
									| NumberFormatException
									| IllegalAccessException e) {

								EdalJettyHandler
										.deleteTicketFromHashMap(ticketAccept);

								this.sendMessage(response,
										EdalJettyHandler.NOT_FOUND_STATUS,
										EdalJettyHandler.NOT_FOUND,
										"Can not to accept ticket it seems to be already accepted/rejected: "
												+ e.getMessage());
								break;
							}
						} else {
							this.sendMessage(response,
									EdalJettyHandler.NOT_FOUND_STATUS,
									EdalJettyHandler.NOT_FOUND,
									"No ReviewerCode definded");
							break;
						}

						this.sendMessage(response, EdalJettyHandler.OK_STATUS,
								"OK", "Thank you");

						break;

					/* reject an object */
					case REJECT:

						final String ticketReject = tokenizer.nextToken();

						if (tokenizer.hasMoreElements()) {

							try {
								final int reviewerHashCode = Integer
										.parseInt(tokenizer.nextToken());

								if (checkIfAllowedToClick(reviewerHashCode,
										ticketReject)) {

									DataManager
											.getImplProv()
											.getApprovalServiceProvider()
											.newInstance()
											.reject(ticketReject,
													reviewerHashCode);
								} else {
									this.sendMessage(response,
											EdalJettyHandler.NOT_FOUND_STATUS,
											EdalJettyHandler.NOT_FOUND, "Already clicked");
									break;
								}

							} catch (EdalApprovalException
									| InstantiationException
									| NumberFormatException
									| IllegalAccessException e) {

								EdalJettyHandler
										.deleteTicketFromHashMap(ticketReject);

								this.sendMessage(response,
										EdalJettyHandler.NOT_FOUND_STATUS,
										EdalJettyHandler.NOT_FOUND,
										"Can not to reject ticket  it seems to be already accepted/rejected: "
												+ e.getMessage());
								break;
							}
						} else {
							this.sendMessage(response,
									EdalJettyHandler.NOT_FOUND_STATUS,
									EdalJettyHandler.NOT_FOUND,
									"No ReviewerCode definded");
							break;
						}

						this.sendMessage(response, EdalJettyHandler.OK_STATUS,
								"OK", "Thank you");

						break;
					/* other method defined */

					case LOGIN:

						final String uuid = tokenizer.nextToken();
						final String emailAddress = tokenizer.nextToken();

						Boolean successful = false;
						try {
							successful = DataManager.getImplProv()
									.validateRootUser(
											new InternetAddress(emailAddress),
											UUID.fromString(uuid));
						} catch (final AddressException e) {
							this.sendMessage(response,
									EdalJettyHandler.NOT_FOUND_STATUS,
									"eDAL-Server Admin confirmation failed", "");
						}

						if (successful) {
							try {
								this.sendMessage(
										response,
										EdalJettyHandler.OK_STATUS,
										"eDAL-Server Admin confirmation success",
										"Thank you <br/>You have successfully registered as administrator for eDAL-Server on "
												+ EdalJettyServer
														.getServerURL()
												+ "<br/>The server is now started in working mode.");
							} catch (final EdalException e) {
								this.sendMessage(response,
										EdalJettyHandler.NOT_FOUND_STATUS,
										"Error", "unable to load server URL");
							}
						} else {
							this.sendMessage(response,
									EdalJettyHandler.NOT_FOUND_STATUS, "Error",
									"root user validation failed");
						}

						break;

					case LOGO:
						if (tokenizer.hasMoreTokens()) {
							final String logoUrl = tokenizer.nextToken();

							if (logoUrl.equalsIgnoreCase("edal_scaled.png")) {
								this.sendLogo(response, "edal_scaled.png");
								break;
							} else if (logoUrl.equalsIgnoreCase("zip_logo.png")) {
								this.sendLogo(response, "zip_logo.png");
								break;
							}
						}
						break;

					default:

						this.sendMessage(response,
								EdalJettyHandler.NOT_FOUND_STATUS,
								EdalJettyHandler.NOT_FOUND,
								"path has to start with '"
										+ EdalHttpFunctions.ACCEPT + "'"
										+ " or '" + EdalHttpFunctions.REJECT
										+ "'" + " or '"
										+ EdalHttpFunctions.REVIEW + "'");
						break;
					}
				} catch (IllegalArgumentException e) {

					this.sendMessage(response,
							EdalJettyHandler.NOT_FOUND_STATUS,
							EdalJettyHandler.NOT_FOUND,
							"path has to start with '"
									+ EdalHttpFunctions.ACCEPT + "'" + " or '"
									+ EdalHttpFunctions.REJECT + "'" + " or '"
									+ EdalHttpFunctions.REVIEW + "'");

				}

			}
		}
		response.flushBuffer();

	}

	private boolean checkIfAllowedToClick(int reviewerHashCode,
			String ticketAccept) {

		List<String> list = hashmap.get(reviewerHashCode);

		if (list != null) {

			if (list.contains(ticketAccept)) {
				return false;
			} else {
				return true;
			}

		} else {
			List<String> newlist = new ArrayList<String>();
			newlist.add(ticketAccept);
			EdalJettyHandler.hashmap.put(reviewerHashCode, newlist);

			return true;
		}
	}

	/**
	 * Function to response a {@link EdalHttpFunctions#DOWNLOAD} request.
	 * 
	 * @param response
	 * @param entity
	 * @param uuid
	 * @param versionNumber
	 * @param internalId
	 * @param identifierType
	 * @param reviewerId
	 * @throws EdalException
	 */
	private void responseDownloadRequest(HttpServletResponse response,
			PrimaryDataEntity entity, String uuid, Long versionNumber,
			String internalId, String identifierType, String reviewerId)
			throws EdalException {

		if (reviewerId == null) {
			entity = DataManager.getPrimaryDataEntityForPersistenIdentifier(
					uuid, versionNumber,
					PersistentIdentifier.valueOf(identifierType));
			if (!entity.isDirectory()) {
				this.sendFile((PrimaryDataFile) entity, versionNumber, response);
			} else {
				this.sendEntityMetaDataForPersistentIdentifier(response,
						entity, versionNumber,
						PersistentIdentifier.valueOf(identifierType),
						internalId);
			}

		} else {
			entity = DataManager.getPrimaryDataEntityForReviewer(uuid,
					versionNumber, internalId, Integer.parseInt(reviewerId));

			if (!entity.isDirectory()) {
				this.sendFile((PrimaryDataFile) entity, versionNumber, response);
			} else {
				this.sendEntityMetaDataForReviewer(response, entity,
						versionNumber, internalId,
						PersistentIdentifier.valueOf(identifierType),
						Integer.parseInt(reviewerId));

			}
		}
	}

	private void responseZipRequest(HttpServletResponse response,
			PrimaryDataEntity entity, String uuid, long versionNumber,
			String internalId, String identifierType, String reviewerId)
			throws EdalException {

		if (reviewerId == null) {
			entity = DataManager.getPrimaryDataEntityForPersistenIdentifier(
					uuid, versionNumber,
					PersistentIdentifier.valueOf(identifierType));

		} else {
			entity = DataManager.getPrimaryDataEntityForReviewer(uuid,
					versionNumber, internalId, Integer.parseInt(reviewerId));
		}

		if (entity.isDirectory()) {

			try {

				response.setHeader("Content-Disposition", "inline; filename=\""
						+ entity.getName() + ".zip" + "\"");

				response.setStatus(EdalJettyHandler.OK_STATUS);

				CountDownLatch countDownLatch = new CountDownLatch(1);

				OutputStream responseBody = response.getOutputStream();
				// // copy entity bytes stream to HTTP stream
				ZipOutputStream zout = new ZipOutputStream(responseBody);

				ZipThread zipThread = new ZipThread(countDownLatch, zout,
						(PrimaryDataDirectory) entity);

				if (executor.isShutdown()) {
					executor = new EdalThreadPoolExcecutor(1, Math.max(1,
							Runtime.getRuntime().availableProcessors()), 10,
							TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(
									10));
				}
				executor.execute(zipThread);

				try {
					countDownLatch.await();
				} catch (InterruptedException e) {
					throw new EdalException("unable to send zip file", e);
				} finally {
					try {
						zout.close();
						responseBody.close();
					} catch (Exception e) {
						DataManager
								.getImplProv()
								.getLogger()
								.error("unable to close zip streams : "
										+ e.getMessage());
					}
				}
			} catch (IOException e) {
				throw new EdalException("unable to send zip file", e);
			}
		}
	}

	private void sendLogo(HttpServletResponse response, String logoName) {

		try {

			InputStream logo = EdalJettyHandler.class
					.getResourceAsStream(logoName);

			int logoSize = logo.available();

			response.setContentType("image/png");

			response.setHeader("Content-Disposition", "attachment; filename=\""
					+ logoName + "\"");
			response.setContentLength(logoSize);

			response.setStatus(EdalJettyHandler.OK_STATUS);

			CountDownLatch countDownLatch = new CountDownLatch(1);

			final OutputStream responseBody = response.getOutputStream();
			// copy entity bytes stream to HTTP stream
			final PipedInputStream httpIn = new PipedInputStream(
					PipedThread.BUFFER_SIZE);
			final PipedOutputStream pipedOut = new PipedOutputStream(httpIn);

			final PipedThread pipedThread = new PipedThread(httpIn,
					responseBody, countDownLatch, logoSize);

			if (executor.isShutdown()) {
				executor = new EdalThreadPoolExcecutor(1, Math.max(1, Runtime
						.getRuntime().availableProcessors()), 10,
						TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
			}

			executor.execute(pipedThread);

			byte[] buffer = new byte[PipedThread.BUFFER_SIZE];
			int len;
			while ((len = logo.read(buffer)) != -1) {
				pipedOut.write(buffer, 0, len);
			}

			pipedOut.flush();

			try {
				countDownLatch.await();
				logo.close();
				pipedOut.close();
				httpIn.close();
			} catch (InterruptedException e) {
				DataManager
						.getImplProv()
						.getLogger()
						.error("unable to wait for sending logo : "
								+ e.getMessage());
			}
			responseBody.close();

		} catch (IOException e) {
			DataManager.getImplProv().getLogger()
					.error("unable to send logo : " + e.getMessage());
		}

	}

	/**
	 * Generate a HTML output with the
	 * {@link de.ipk_gatersleben.bit.bi.edal.primary_data.metadata.MetaData} of
	 * the given {@link PrimaryDataEntity} and send the output over HTTP.
	 * 
	 * @param response
	 *            the corresponding HTTP exchange.
	 * @param entity
	 *            the the corresponding {@link PrimaryDataEntity}.
	 * @throws EdalException
	 *             if unable to send metadata
	 */
	private void sendEntityMetaData(final HttpServletResponse response,
			final PrimaryDataEntity entity) throws EdalException {

		try {

			final VeloCityHtmlGenerator velo = new VeloCityHtmlGenerator();

			StringWriter w = null;
			if (entity.isDirectory()) {
				w = velo.generateHtmlForDirectory(
						(PrimaryDataDirectory) entity,
						EdalJettyServer.getServerURL());
			} else {
				w = velo.generateHtmlForFile((PrimaryDataFile) entity,
						EdalJettyServer.getServerURL());
			}

			ByteArrayInputStream bis = new ByteArrayInputStream(String.valueOf(
					w.getBuffer()).getBytes());

			response.setContentType("text/html");
			response.setStatus(EdalJettyHandler.OK_STATUS);
			final OutputStream responseBody = response.getOutputStream();

			CountDownLatch countDownLatch = new CountDownLatch(1);

			final PipedInputStream httpIn = new PipedInputStream(
					PipedThread.BUFFER_SIZE);
			final PipedOutputStream pipedOut = new PipedOutputStream(httpIn);

			final PipedThread pipedThread = new PipedThread(httpIn,
					responseBody, countDownLatch, bis.available());

			if (executor.isShutdown()) {
				executor = new EdalThreadPoolExcecutor(1, Math.max(1, Runtime
						.getRuntime().availableProcessors()), 10,
						TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
			}
			executor.execute(pipedThread);

			byte[] buffer = new byte[PipedThread.BUFFER_SIZE];
			int length;
			while ((length = bis.read(buffer)) != -1) {
				pipedOut.write(buffer, 0, length);
			}

			try {
				countDownLatch.await();
			} catch (InterruptedException e) {
				throw new EdalException("unable to send html page", e);
			} finally {
				try {
					bis.close();
					pipedOut.close();
					httpIn.close();
					responseBody.close();
				} catch (Exception e) {
					DataManager
							.getImplProv()
							.getLogger()
							.error("unable to close html streams : "
									+ e.getMessage());
				}
			}
		} catch (IOException e) {
			throw new EdalException("unable to send the html page over HTTP", e);
		} catch (EdalException e) {
			throw e;
		}

	}

	/**
	 * Generate a HTML output with the
	 * {@link de.ipk_gatersleben.bit.bi.edal.primary_data.metadata.MetaData} of
	 * the given {@link PrimaryDataEntity} and send the output over HTTP.
	 * 
	 * @param response
	 *            the corresponding HTTP exchange.
	 * @param entity
	 *            the the corresponding {@link PrimaryDataEntity}.
	 * @param identifierType
	 * @param versionNumber
	 *            the current
	 *            {@link de.ipk_gatersleben.bit.bi.edal.primary_data.file.PrimaryDataEntityVersion}
	 *            .
	 * @throws EdalException
	 *             if unable to send metadata
	 */
	private void sendEntityMetaDataForPersistentIdentifier(
			final HttpServletResponse response, final PrimaryDataEntity entity,
			final long versionNumber,
			final PersistentIdentifier identifierType, final String internalId)
			throws EdalException {

		try {
			final VeloCityHtmlGenerator velo = new VeloCityHtmlGenerator();

			StringWriter w = null;
			if (entity.isDirectory()) {
				w = velo.generateHtmlForDirectoryOfSnapshot(
						(PrimaryDataDirectory) entity, versionNumber,
						internalId, identifierType,
						EdalJettyServer.getServerURL());
			} else {
				w = velo.generateHtmlForFileOfSnapshot(
						(PrimaryDataFile) entity, versionNumber,
						identifierType, EdalJettyServer.getServerURL(),
						internalId);
			}

			ByteArrayInputStream bis = new ByteArrayInputStream(String.valueOf(
					w.getBuffer()).getBytes());

			response.setContentType("text/html");
			response.setStatus(EdalJettyHandler.OK_STATUS);
			final OutputStream responseBody = response.getOutputStream();

			CountDownLatch countDownLatch = new CountDownLatch(1);

			final PipedInputStream httpIn = new PipedInputStream(
					PipedThread.BUFFER_SIZE);
			final PipedOutputStream pipedOut = new PipedOutputStream(httpIn);

			final PipedThread pipedThread = new PipedThread(httpIn,
					responseBody, countDownLatch, bis.available());

			if (executor.isShutdown()) {
				executor = new EdalThreadPoolExcecutor(1, Math.max(1, Runtime
						.getRuntime().availableProcessors()), 10,
						TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
			}
			executor.execute(pipedThread);

			byte[] buffer = new byte[PipedThread.BUFFER_SIZE];
			int length;
			while ((length = bis.read(buffer)) != -1) {
				pipedOut.write(buffer, 0, length);
			}

			try {
				countDownLatch.await();
			} catch (InterruptedException e) {
				throw new EdalException("unable to send html page", e);
			} finally {
				try {
					bis.close();
					pipedOut.close();
					httpIn.close();
					responseBody.close();
				} catch (Exception e) {
					DataManager
							.getImplProv()
							.getLogger()
							.error("unable to close html streams : "
									+ e.getMessage());
				}
			}
		} catch (IOException e) {
			throw new EdalException("unable to send the html page over HTTP", e);
		} catch (EdalException e) {
			throw e;
		}
	}

	/**
	 * Generate a HTML output with the
	 * {@link de.ipk_gatersleben.bit.bi.edal.primary_data.metadata.MetaData} of
	 * the given {@link PrimaryDataEntity} and send the output over HTTP. This
	 * function is especially for the Output of the landing page for a reviewer,
	 * who should approve the requested
	 * {@link de.ipk_gatersleben.bit.bi.edal.primary_data.file.PublicReference}.
	 * 
	 * @param response
	 *            the corresponding HTTP exchange.
	 * @param entity
	 *            the the corresponding {@link PrimaryDataEntity}.
	 * @param reviewerCode
	 *            the reviewerCode to identify a reviewer
	 * @throws EdalException
	 *             if unable to send metadata
	 */
	private void sendEntityMetaDataForReviewer(
			final HttpServletResponse response, final PrimaryDataEntity entity,
			final long versionNumber, final String internalId,
			final PersistentIdentifier identifierType, final int reviewerCode)
			throws EdalException {
		try {

			final VeloCityHtmlGenerator velo = new VeloCityHtmlGenerator();

			StringWriter w = null;
			if (entity.isDirectory()) {
				w = velo.generateHtmlForDirectoryForReviewer(
						(PrimaryDataDirectory) entity, versionNumber,
						internalId, identifierType,
						EdalJettyServer.getServerURL(), reviewerCode);
			} else {
				w = velo.generateHtmlForFileForReviewer(
						(PrimaryDataFile) entity, versionNumber, internalId,
						identifierType, EdalJettyServer.getServerURL(),
						reviewerCode);
			}

			ByteArrayInputStream bis = new ByteArrayInputStream(String.valueOf(
					w.getBuffer()).getBytes());

			response.setContentType("text/html");
			response.setStatus(EdalJettyHandler.OK_STATUS);
			final OutputStream responseBody = response.getOutputStream();

			CountDownLatch countDownLatch = new CountDownLatch(1);

			final PipedInputStream httpIn = new PipedInputStream(
					PipedThread.BUFFER_SIZE);
			final PipedOutputStream pipedOut = new PipedOutputStream(httpIn);

			final PipedThread pipedThread = new PipedThread(httpIn,
					responseBody, countDownLatch, bis.available());

			if (executor.isShutdown()) {
				executor = new EdalThreadPoolExcecutor(1, Math.max(1, Runtime
						.getRuntime().availableProcessors()), 10,
						TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
			}
			executor.execute(pipedThread);

			byte[] buffer = new byte[PipedThread.BUFFER_SIZE];
			int length;
			while ((length = bis.read(buffer)) != -1) {
				pipedOut.write(buffer, 0, length);
			}

			try {
				countDownLatch.await();
			} catch (InterruptedException e) {
				throw new EdalException("unable to send html page", e);
			} finally {
				try {
					bis.close();
					pipedOut.close();
					httpIn.close();
					responseBody.close();
				} catch (Exception e) {
					DataManager
							.getImplProv()
							.getLogger()
							.error("unable to close html streams : "
									+ e.getMessage());
				}
			}
		} catch (IOException e) {
			throw new EdalException("unable to send the html page over HTTP", e);
		} catch (EdalException e) {
			throw e;
		}
	}

	/**
	 * Send the data of the corresponding {@link PrimaryDataFile} to the HTTP
	 * {@link OutputStream}.
	 * 
	 * @param file
	 *            the corresponding {@link PrimaryDataFile} to send.
	 * @param response
	 * @throws EdalException
	 *             if unable to send {@link PrimaryDataFile}
	 */
	private void sendFile(final PrimaryDataFile file, final long versionNumber,
			final HttpServletResponse response) throws EdalException {
		final PrimaryDataFile currentFile = file;
		try {
			currentFile.switchCurrentVersion(currentFile
					.getVersionByRevisionNumber(versionNumber));
		} catch (PrimaryDataEntityVersionException e1) {
			this.sendMessage(response, EdalJettyHandler.NOT_FOUND_STATUS,
					EdalJettyHandler.NOT_FOUND, "no file found");
		}

		try {
			String type = "";
			Long size = null;
			try {
				type = ((DataFormat) currentFile.getCurrentVersion()
						.getMetaData()
						.getElementValue(EnumDublinCoreElements.FORMAT))
						.getMimeType();
				size = ((DataSize) currentFile.getCurrentVersion()
						.getMetaData()
						.getElementValue(EnumDublinCoreElements.SIZE))
						.getFileSize();

			} catch (final MetaDataException e) {
				throw new EdalException(
						"unable to load the MIME type/file size", e);
			}

			response.setContentType(type);
			response.setHeader("Content-Disposition", "inline; filename=\""
					+ currentFile.getName() + "\"");
			response.setContentLength(size.intValue());

			response.setStatus(EdalJettyHandler.OK_STATUS);

			CountDownLatch countDownLatch = new CountDownLatch(1);

			final OutputStream responseBody = response.getOutputStream();
			// copy entity bytes stream to HTTP stream
			final PipedInputStream httpIn = new PipedInputStream(
					PipedThread.BUFFER_SIZE);
			final PipedOutputStream pipedOut = new PipedOutputStream(httpIn);

			final PipedThread pipedThread = new PipedThread(httpIn,
					responseBody, countDownLatch, size);

			if (executor.isShutdown()) {
				executor = new EdalThreadPoolExcecutor(1, Math.max(1, Runtime
						.getRuntime().availableProcessors()), 10,
						TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
			}
			executor.execute(pipedThread);

			currentFile.read(pipedOut);

			try {
				countDownLatch.await();
			} catch (InterruptedException e) {
				throw new EdalException("unable to send file", e);
			} finally {
				try {
					pipedOut.close();
					httpIn.close();
					responseBody.close();
				} catch (Exception e) {
					DataManager
							.getImplProv()
							.getLogger()
							.error("unable to close download streams : "
									+ e.getMessage());
				}
			}
		} catch (final PrimaryDataFileException | IOException e) {
			throw new EdalException("unable to send file", e);
		}
	}

	/**
	 * Send a response with the given message and responseCode.
	 * 
	 * @param response
	 *            the corresponding HTTP exchange.
	 * @param responseCode
	 *            the response code for this message (e.g. 200,404...).
	 * @param title
	 *            the title for the responseCode(e.g. OK, Not Found...).
	 * @param message
	 *            the message to send.
	 */
	private void sendMessage(final HttpServletResponse response,
			final int responseCode, final String title, final String message) {

		try {

			final VeloCityHtmlGenerator velo = new VeloCityHtmlGenerator();

			final String htmlOutput = velo.generateHtmlForErrorMessage(
					responseCode, title, message,
					EdalJettyServer.getServerURL()).toString();

			response.setStatus(responseCode);

			response.setContentType("text/html");

			final OutputStream responseBody = response.getOutputStream();
			responseBody.write(htmlOutput.getBytes());
			responseBody.close();
		} catch (IOException | EdalException e) {

			e.printStackTrace();
			DataManager
					.getImplProv()
					.getLogger()
					.error("unable to send " + responseCode + "-message : "
							+ e.getMessage());
		}
	}

	public static void deleteTicketFromHashMap(String ticket) {

		for (Entry<Integer, List<String>> entry : hashmap.entrySet()) {

			List<String> list = entry.getValue();

			if (list.contains(ticket)) {
				list.remove(ticket);
			}
		}
	}
}